package cn.doitedu.cn.doitedu.rtdw.etl;

import cn.doitedu.cn.doitedu.rtdw.pojo.PageStayLong;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

public class Job3_PageStayTimeLongOlapSupport {

    public static void main(String[] args) throws Exception {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 用source读dwd层的kafka行为明细数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events-detail")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setGroupId("doit40-c-1")
                .build();

        DataStreamSource<String> jsonStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");
        SingleOutputStreamOperator<PageStayLong> beanStream = jsonStream.map(json -> JSON.parseObject(json, PageStayLong.class));


        SingleOutputStreamOperator<PageStayLong> processed = beanStream.keyBy(PageStayLong::getSession_id)
                .process(new KeyedProcessFunction<String, PageStayLong, PageStayLong>() {
                    ValueState<PageStayLong> beanState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 开启3小时的ttl，并且只要state被读或写，都要更新ttl的计时
                        ValueStateDescriptor<PageStayLong> desc = new ValueStateDescriptor<>("bean", PageStayLong.class);
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(3))
                                .updateTtlOnReadAndWrite()
                                .build();
                        desc.enableTimeToLive(ttlConfig);

                        beanState = getRuntimeContext().getState(desc);
                    }

                    /* *  *  *  * *核心逻辑  * *  * *  * *  *  */
                    // 1. 填充字段：页面open时间
                    // 2. 遇到 pageLoad,输出上一个页面的封闭事件，页面open是上一个页面的，页面url也是上一个页面的，事件时间用最新时间
                    // 2. 遇到 wakeup, 更新所在页面的open时间
                    @Override
                    public void processElement(PageStayLong pageStayLong, KeyedProcessFunction<String, PageStayLong, PageStayLong>.Context context, Collector<PageStayLong> collector) throws Exception {

                        PageStayLong prePageStayLong = beanState.value();
                        if (prePageStayLong == null) {
                            // 填充新数据的openTime
                            pageStayLong.setPageOpenTime(pageStayLong.getEvent_time());
                            // 并覆盖掉状态中的信息
                            beanState.update(pageStayLong);
                        }


                        if (pageStayLong.getEvent_id().equals("page_load")) {

                            // 把上一个页面信息中的eventTime更新成最新的eventTime
                            prePageStayLong.setEvent_time(pageStayLong.getEvent_time());
                            // 输出封闭事件
                            collector.collect(prePageStayLong);

                            // 填充新数据的openTime
                            pageStayLong.setPageOpenTime(pageStayLong.getEvent_time());
                            // 并覆盖掉状态中的信息
                            beanState.update(pageStayLong);


                        } else if ("wake_up".equals(pageStayLong.getEvent_id())) {

                            // 填充本条数据的openTime
                            pageStayLong.setPageOpenTime(pageStayLong.getEvent_time());

                            // 更新掉状态中的数据
                            beanState.update(pageStayLong);

                        }


                        // 将状态中的数据，修改掉eventTime，输出
                        PageStayLong value = beanState.value();
                        value.setEvent_time(pageStayLong.getEvent_time());
                        collector.collect(value);


                        // 状态清理，如果收到了appClose事件，则会话明确结束，可以清理掉状态
                        if("app_close".equals(pageStayLong.getEvent_id())){
                            beanState.clear();
                        }

                    }
                });

        // 将做好虚拟插值事件的数据流，转成表，来做后续的统计
        tenv.createTemporaryView("processed", processed,
                Schema.newBuilder()
                        .column("user_id", DataTypes.BIGINT())
                        .column("session_id", DataTypes.STRING())
                        .column("event_time", DataTypes.BIGINT())
                        .column("page_url", DataTypes.STRING())
                        .column("pageOpenTime", DataTypes.BIGINT())
                        .columnByExpression("rt", "to_timestamp_ltz(event_time,3)")
                        .watermark("rt", "rt - interval '0' second")
                        .build());



        // 创建doris映射表，来接收统计结果
        /**
         *     dt    DATEV2,
         *     user_id  BIGINT,
         *     session_id  VARCHAR(20),
         *     page_url  VARCHAR(20),
         * 	   page_open_time BIGINT,
         *     stay_time_long BIGINT  SUM
         */

        tenv.executeSql(
                " CREATE TABLE acc_timelong_page_doris(           "+
                        "          dt                      DATE             "+
                        "         ,user_id                 bigint           "+
                        "         ,session_id              string           "+
                        "         ,page_url                string           "+
                        "         ,page_open_time          bigint           "+
                        "         ,stay_time_long          bigint           "+
                        " ) WITH (                                          "+
                        "    'connector' = 'doris',                         "+
                        "    'fenodes' = 'doitedu:8030',                    "+
                        "    'table.identifier' = 'dws.acc_timelong_page', "+
                        "    'username' = 'root',                           "+
                        "    'password' = 'root',                           "+
                        "    'sink.label-prefix' = 'doris_label-uuu'        "+
                        " )                                                 "
        );


        // 写sql，按5分钟滚动窗口，来统计每个页面每一次停留的时长
        // 一个页面的一次停留中的一组数据，可能被划分到两个不同的滚动时间窗口，则该次停留时长会被输出2条结果
        // 但是，我们可以子doris中利用聚合表模型，进行合并
        tenv.executeSql("create temporary view filtered as select * from processed where page_url is not null");

        tenv.executeSql(
                "insert into  acc_timelong_page_doris " +
                        "SELECT                                                           " +
                        " 	to_date(date_format(window_end,'yyyy-MM-dd')) as dt,           " +
                        " 	user_id,                                                       " +
                        " 	session_id,                                                    " +
                        " 	page_url,                                                      " +
                        " 	pageOpenTime,                                                  " +
                        " 	max(event_time) - min(event_time) as stay_time_long            " +
                        " FROM TABLE(                                                      " +
                        "     TUMBLE(TABLE filtered,DESCRIPTOR(rt),INTERVAL '5' MINUTE)    " +
                        " )                                                                " +
                        " GROUP BY                                                         " +
                        "     window_start,                                                " +
                        " 	window_end,                                                    " +
                        " 	user_id,                                                       " +
                        " 	session_id,                                                    " +
                        " 	page_url,                                                      " +
                        " 	pageOpenTime                                                   "

        );


        env.execute();

    }
}
